package com.dianziq.im.server;

import java.io.File;
import java.nio.charset.Charset;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.log4j.Logger;
import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.buffer.ChannelBuffers;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.ChannelFutureListener;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ExceptionEvent;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelHandler;
import org.jboss.netty.handler.codec.http.Cookie;
import org.jboss.netty.handler.codec.http.CookieDecoder;
import org.jboss.netty.handler.codec.http.DefaultHttpResponse;
import org.jboss.netty.handler.codec.http.HttpHeaders;
import org.jboss.netty.handler.codec.http.HttpHeaders.Names;
import org.jboss.netty.handler.codec.http.HttpHeaders.Values;
import org.jboss.netty.handler.codec.http.HttpRequest;
import org.jboss.netty.handler.codec.http.HttpResponseStatus;
import org.jboss.netty.handler.codec.http.HttpVersion;
import org.jboss.netty.handler.codec.http.QueryStringDecoder;
import org.jboss.netty.util.HashedWheelTimer;
import com.dianziq.im.server.cmd.Action;
import com.dianziq.im.server.cmd.Context;

public class ConnectionService extends SimpleChannelHandler {
	private static final Logger log=Logger.getLogger(ConnectionService.class);
	static Charset UTF8 = Charset.forName("UTF-8");
	//存放每个用户所产生的ConnectionBag，包含了连接通道Channel 和计时器
	ConcurrentHashMap<Integer, ConnectionBag> conn = new ConcurrentHashMap<Integer, ConnectionBag>();
	HashedWheelTimer timer = new HashedWheelTimer();
	//Executors、ExecutorService 以及 CompletionService 线程工具类
	
	java.util.concurrent.ExecutorService sendMsgPool = java.util.concurrent.Executors
			.newFixedThreadPool(1);

	//内部类
	public class ConnectionBag  {
		/**
		 * 　ConcurrentSkipListMap有几个ConcurrentHashMap 不能比拟的优点：
		 
		　　1、ConcurrentSkipListMap 的key是有序的。
		 
		　　2、ConcurrentSkipListMap 支持更高的并发。ConcurrentSkipListMap 的存取时间是log（N），和线程数几乎无关。也就是说在数据量一定的情况下，并发的线程越多，ConcurrentSkipListMap越能体现出他的优势。
		http://www.holdjava.com/threadinfo/85032.htm
		 */
		final ConcurrentSkipListMap<Integer, Channel> channels = new java.util.concurrent.ConcurrentSkipListMap<Integer, Channel>();
		//Timeout timeout;
		Status status=Status.Online;
		//原子性的int 支持并非
		final AtomicInteger index = new AtomicInteger();
		volatile int lastRespTime;
		volatile int emptyStartTime = Integer.MAX_VALUE;

		public void addChannel(Channel c, int fromIndex, final int key) {
			c.getCloseFuture().addListener(new ChannelFutureListener() {
				@Override
				public void operationComplete(ChannelFuture arg0)
						throws Exception {
					//key 和value 都相等的时候才
					channels.remove(key, arg0.getChannel());
					if (channels.size() == 0) {
						//关闭通道的时候设置当前时间，方便提醒好友自己已经下线
						emptyStartTime = now();
					}

				}

			});
			this.channels.put(key, c);
			emptyStartTime = Integer.MAX_VALUE;
			this.lastRespTime = now();

		}

		private int now() {
			return (int) (System.currentTimeMillis() / 1000);
		}

		public void check(int time, int uid) {
			if (time - this.emptyStartTime > 3) {
				conn.remove(uid, this);
				onStatusChang(uid, Status.Offline);
				return;
			}
			//30秒就断开链接，客户端重新发起
			if (time - this.lastRespTime > 30) {
				this.write(Message.HEARTBEAT);
			}

		}

		public void write(Message s) {
			int idx = this.index.incrementAndGet();
			// this.lastMsg = s;
			DefaultHttpResponse resp = new DefaultHttpResponse(
					HttpVersion.HTTP_1_1, HttpResponseStatus.OK);
			 
			resp.setHeader(Names.CONTENT_TYPE, "text/txt; charset=UTF-8");
			String ss="IM.onMsg("+s.toString(idx)+")";
			ChannelBuffer cb = ChannelBuffers.wrappedBuffer(ss.getBytes(UTF8));
			resp.setContent(cb);
			resp.setHeader(Names.CONNECTION, Values.KEEP_ALIVE);
			resp.setHeader(Names.CONTENT_LENGTH, cb.readableBytes());

			for (Channel c : this.channels.values()) {

				try {
					if (c.isConnected())
						c.write(resp);// .addListener(ChannelFutureListener.CLOSE);
				} catch (Exception e) {
					log.info(e);
				}
			}
		}

		 

	}
	//维护链接线程。超时就返回，客户端重新发起请求
	public ConnectionService() {
		this.timer.start();
		new Thread() {
			public void run() {
				while (true) {

					int i = (int) (System.currentTimeMillis() / 1000);
					for (Map.Entry<Integer, ConnectionBag> e : conn.entrySet()) {
						try {
							e.getValue().check(i, e.getKey());
						} catch (Exception ew) {
							ew.printStackTrace();
						}
					}
					try {
						Thread.sleep(1000);
					} catch (InterruptedException e) {
						return;
					}
				}
			}

		}.start();
	}

	public void messageReceived(ChannelHandlerContext ctx, final MessageEvent me)
			throws Exception {
		HttpRequest hr = (HttpRequest) me.getMessage();
		QueryStringDecoder queryStringDecoder = new QueryStringDecoder(
				hr.getUri(), UTF8);
		String path = queryStringDecoder.getPath();
		Map<String, List<String>> params = queryStringDecoder.getParameters();
		Channel ch = me.getChannel();
		CookieDecoder cd = new CookieDecoder();
		String ck = hr.getHeader(HttpHeaders.Names.COOKIE);

		Set<Cookie> cookies = cd.decode(ck == null ? "" : ck);
		//Channel ch, ConnectionService service, String path,Map<String, List<String>> params, Set<Cookie> cookies
		Context c = new Context(ch, this, path, params, cookies);
		//这边就初始化了Acetion 内的数据，static 是编译的时候生产
		Action cmd = Action.from(path);
		if (cmd != null) {
			cmd.handle(c);
		} else {
		//	if (path.startsWith("/static/")) {
				File f = new File("D:/svnroot/im/web"+path );
				if (f.exists()) {

					c.serveStatic(f);
					return;
				}
			 
			c.sendError("Cannot find :" + path);

		}
	}
 

	public ConnectionBag get(int userId) {
		ConnectionBag cb = new ConnectionBag();

		ConnectionBag back = this.conn.putIfAbsent(userId, cb);
		if (back == null) {
			back = cb;
			this.onStatusChang(userId,Status.Online );
		}
		return back;
	}

	private List<Integer> getFriends(int uid) throws SQLException {
		List<Integer> ret = DBBag.getList("select destuid from friend where u_id="+uid,
				new DBBag.Mapper<Integer>() {

					@Override
					public Integer map(ResultSet res) throws SQLException {
						// TODO Auto-generated method stub
						return res.getInt(1);
					}
				});

		return ret;
	}
	public Status status(int userId){
		ConnectionBag cb= this.conn.get(userId);
		if(cb==null){
			return Status.Offline;
		}
		if(cb.status==null){
			return Status.Offline;
		}
		return cb.status;
	}

	/**
	 * status might be :login ,logout,idle etc.
	 * 
	 * @param uid
	 * @param status
	 */
	private void onStatusChang(int uid, Status status) { 
		Message m = new Message();
		m.fromUID = uid;
		m.type = Message.MsgType.Control;
		m.content = new StringBuilder("{fun:'IM.OSC(").append( uid).append(",").append(status.ordinal()).append(")'}").toString();

		try {
			List <Integer> fl=this.getFriends(uid);
			if(fl.size()==0){
				return;
			}
			for (int i : fl) {
				this.sendMsg(i, m);
			}
		} catch (SQLException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	public void onCmd(int uid, Channel c,  String msg) {
		Message m = new Message();
		m.fromUID = uid;
		m.type = Message.MsgType.Control;
		m.content = msg;

		this.sendMsg(uid, m);
	}

	public void onMsg(int uid, int touid, Channel c, String msg) {
		Message m = new Message();
		m.fromUID = uid;
		m.toUID=touid;
		m.type = Message.MsgType.Default;
		m.content = msg;

		this.sendMsg(uid, m);
		this.sendMsg(touid, m);

	}

	public void sendMsg(final int destuid, final Message msg) {
		// ConnectionBag cbag = conn.get(destuid);
		// if (cbag != null)
		// cbag.write(msg);
		//
		// //
		this.sendMsgPool.submit(new Runnable() {

			@Override
			public void run() {
				ConnectionBag cbag = conn.get(destuid);

				cbag.write(msg);

			}

		});

	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e)
			throws Exception {
		e.getCause().printStackTrace();

		e.getChannel().close();
	}


	
}
